package org.shj.hadoop.customer;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RunJob {
	static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	static class TemperatureMapper extends Mapper<LongWritable, Text, KeyPair, Text>{
		
		//每个mapper执行一次
		protected void setup(Mapper<LongWritable, Text, KeyPair, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			super.setup(context);
		}

		//每个mapper执行多次
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, KeyPair, Text>.Context context)
				throws IOException, InterruptedException {
			try{
				String line = value.toString();
				String[] ss = line.split(" ");
				if(ss.length >= 3 ){
//					Calendar cal = Calendar.getInstance();
//					cal.setTime(sdf.parse(ss[0]));
//					int year = cal.get(Calendar.YEAR);
					int year = Integer.parseInt(ss[0].substring(0, 4));
					int temperature = Integer.parseInt(ss[2]);
					KeyPair kp = new KeyPair();
					kp.setYear(year);
					kp.setTemperature(temperature);
					
					context.write(kp, value);
					
				}
			}catch(Exception e){
				e.printStackTrace();
			}
		}
		
	}
	
	static class TemperatureReduce extends Reducer<KeyPair, Text, KeyPair, Text>{

		@Override
		protected void reduce(KeyPair key, Iterable<Text> values, Reducer<KeyPair, Text, KeyPair, Text>.Context context)
				throws IOException, InterruptedException {
			for(Text t : values){
				context.write(key, t);
			}
		}
		
	}
	
	public static void main(String[] args){
		Configuration conf = new Configuration();
		try{
			Job job = Job.getInstance(conf, "temperature");
			job.setJarByClass(RunJob.class);
			job.setMapperClass(TemperatureMapper.class);
			job.setReducerClass(TemperatureReduce.class);
			
			job.setMapOutputKeyClass(KeyPair.class);
			job.setMapOutputValueClass(Text.class);
			
			job.setNumReduceTasks(4);
			job.setGroupingComparatorClass(GroupTemperature.class);
			job.setPartitionerClass(CustomerPartition.class);
			job.setSortComparatorClass(SortTemperature.class);
			
			FileInputFormat.addInputPath(job, new Path("/shj/input/temperature/"));
			FileOutputFormat.setOutputPath(job, new Path("/shj/output/temperature/"));
			System.exit(job.waitForCompletion(true) ? 0 : 1);
			
		}catch(Exception e){
			e.printStackTrace();
		}
	}
}
